package com.ruoyi.mqtt.client;

import com.ruoyi.mqtt.callback.DefaultCallback;
import com.ruoyi.mqtt.config.MqttConfig;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

/**
 * Mqtt 创建连接 消息发布 订阅
 * 第一家中继器连接类
 */
@Component
public class MyMqttClient {

    private static final Logger logger = LoggerFactory.getLogger(MyMqttClient.class);
    @Autowired
    private MqttConfig config;

    @Autowired
    private DefaultCallback defaultCallback;

    private org.eclipse.paho.client.mqttv3.MqttClient client;
    private static MemoryPersistence memoryPersistence = null;
    /** options MQTT Cleint连接配置 */
    private MqttConnectOptions options;

    public org.eclipse.paho.client.mqttv3.MqttClient getClient() {
        return client;
    }

    public void setClient(org.eclipse.paho.client.mqttv3.MqttClient client) {
        this.client = client;
    }

    /**
     * connect : 客户端连接
     */
    @PostConstruct
    public void connect() {
        logger.info("初始化连接");
        try {
            //            设置持久化方式
            memoryPersistence = new MemoryPersistence();
            this.client = new org.eclipse.paho.client.mqttv3.MqttClient(config.getUrls(), config.getClientId(), memoryPersistence);
            options = new MqttConnectOptions();
            options.setCleanSession(true);
            options.setUserName(config.getUsername());
            options.setPassword(config.getPassword().toCharArray());
            options.setConnectionTimeout(config.getCompletionTimeout());
            options.setKeepAliveInterval(30);
            // 重连清空session
            options.setAutomaticReconnect(true);
            client.setCallback(defaultCallback);
            if (null !=client){
                client.connect(options);
            }
        } catch (MqttException e) {
            logger.error(e.getCause().getLocalizedMessage());
            e.printStackTrace();
        }
    }

    /**
     * publish : 发布消息
     * 向某个主题发送消息
     * @param qos         连接方式
     * @param retained    是否保留
     * @param topic       主题
     * @param pushMessage 消息体
     */
    public void publish(int qos, boolean retained, String topic, String pushMessage) {
        MqttMessage message = new MqttMessage();
        message.setQos(qos);
        message.setRetained(retained);
        message.setPayload(pushMessage.getBytes());
        MqttTopic mTopic = this.getClient().getTopic(topic);
        if (null == mTopic) {
            logger.info("topic not exist");
        }
         MqttDeliveryToken token;
        try {
            token = mTopic.publish(message);
            token.waitForCompletion();
        } catch (MqttPersistenceException e) {
            e.printStackTrace();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 订阅某个主题
     * @param topic 主题
     * @param qos 消息质量 0 1 2
     */
    public void subscribe(String topic, int qos) {
        if (null != client && client.isConnected()) {
            logger.info("开始订阅主题" + topic);
            try {
                this.getClient().subscribe(topic, qos);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }else {
            System.out.println("mqttClient is error");
        }
    }

    /**
     * 取消订阅
     * @param topic 要取消的主题
     */
    public void cleanTopic(String topic) {
        if (null != client && !client.isConnected()) {
            try {
                client.unsubscribe(topic);
            } catch (MqttException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        } else {
            System.out.println("mqttClient is error");
        }
    }

    /**
     * 关闭连接
     */
    @PreDestroy
    public void closeConnect() {
        //关闭存储方式
        if (null != memoryPersistence) {
            try {
                memoryPersistence.close();
            } catch (MqttPersistenceException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        } else {
            logger.error("memoryPersistence is null");
        }

//        关闭连接
        if (null != client) {
            if (client.isConnected()) {
                try {
                    client.disconnect();
                    client.close();
                } catch (MqttException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                    logger.error("关闭连接异常={}",e.getMessage());
                }
            } else {
                logger.error("关闭连接异常={}","mqttClient is not connect");
            }
        } else {
            logger.error("关闭连接异常={}","mqttClient is null");
        }
    }
}
